package com.example.hbase;

import com.alibaba.fastjson.JSON;
import com.example.hbase.config.HBaseConfig;
import com.example.hbase.model.Product;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.util.Bytes;

import javax.servlet.ServletException;
import javax.servlet.http.HttpServlet;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.io.IOException;
import java.util.Map;
import java.util.TreeMap;

//@WebServlet ("/HBaseServlet")
public class HBaseServlet extends HttpServlet {
    //top5例子（两种写法）
    private static final String TABLE_NAME = "products";
    private static final String COLUMN_FAMILY = "info";
    private static final String[] QUALIFIERS = {"name", "rating"};

    protected void doGet(HttpServletRequest request, HttpServletResponse response) throws ServletException, IOException {


        Connection connection = null;
        Table table = null;
        // Use TreeMap to sort by rating (descending) 使用treemap排序rating列（倒序）
        TreeMap<Float, Product> productMap = new TreeMap<>(new java.util.Comparator<Float>() {
            public int compare(Float r1, Float r2) {
                return r2.compareTo(r1); // Descending order（倒序）
            }
        });

        try {
            // Initialize HBase connection 初始化hbase链接
            connection = HBaseConfig.getConnection();
            System.out.println(connection);
            table = connection.getTable(TableName.valueOf(TABLE_NAME));

            // Create filter for rating > 0（筛选rating>0)
            SingleColumnValueFilter filter = new SingleColumnValueFilter(
                    Bytes.toBytes(COLUMN_FAMILY),
                    Bytes.toBytes("rating"),
                    CompareFilter.CompareOp.GREATER,
                    Bytes.toBytes(0.0f)
            );
            filter.setFilterIfMissing(true); // Skip rows without rating column(如果没有rating列就跳过）

            // Scan with filter （从筛选器读取）
            Scan scan = new Scan();
            scan.addFamily(Bytes.toBytes(COLUMN_FAMILY));
            scan.setFilter(filter);
            ResultScanner scanner = table.getScanner(scan);

            // Retrieve and store in TreeMap （查询和存储到treemap）
            for (Result result : scanner) {
                String rowKey = Bytes.toString(result.getRow());
                String name = Bytes.toString(result.getValue(Bytes.toBytes(COLUMN_FAMILY), Bytes.toBytes("name")));
                String ratingStr = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("rating")));
                float rating = Float.parseFloat(ratingStr);
//                TreeMap.put() uses the comparator to maintain sorted order.
//                If duplicate ratings exist, the last Product overwrites
//                earlier ones (no duplicates by key in a Map).
                // TreeMap.put() 使用比较器来帮助排序。
                // 如果存在重复的rating，则将最后一个Product
                // 覆盖较早的Product（Map 中不存在键重复的product）。
                productMap.put(rating, new Product(rowKey, name, rating));
            }
            scanner.close();

            // Get top 5 products 读取top5的product

//          Creates an array to hold up to 5 Product objects. 创建一个数组保存5个product的对象
            Product[] topProducts = new Product[Math.min(5, productMap.size())];

            int index = 0;
            for (Map.Entry<Float, Product> entry : productMap.entrySet()) {
                if (index < 5) {
                    topProducts[index] = entry.getValue();
                    index++;
                } else {
                    break;
                }
            }
//            System.out.println(topProducts);
            // Convert to JSON and forward to JSP 转换到JSON并且传到JSP
            String jsonData = JSON.toJSONString(topProducts);
//            System.out.println("JSON Data: " + jsonData);
            request.setAttribute("productData", jsonData);
            request.getRequestDispatcher("/index.jsp").forward(request, response);

        } catch (IOException e) {
            e.printStackTrace();
            response.sendError(HttpServletResponse.SC_INTERNAL_SERVER_ERROR, "Error accessing HBase");
        } finally {
            if (table != null) table.close();
            if (connection != null) connection.close();
        }
    }
}